package com.atguigu.flink.chapter07;

import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.streaming.api.datastream.BroadcastConnectedStream;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

/**
 * TODO 广播状态
 *
 * @author cjp
 * @version 1.0
 * @date 2021/1/26 13:52
 */
public class Flink29_State_BroadcastState {

    static final MapStateDescriptor<String, String> broadcastStateDesc = new MapStateDescriptor<>("broadcastState", Types.STRING, Types.STRING);

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<String> inputDS = env.socketTextStream("localhost", 7777);
        DataStreamSource<String> controlDS = env.socketTextStream("localhost", 8888);

        // TODO 1.将流广播出去
        BroadcastStream<String> controlBS = controlDS.broadcast(broadcastStateDesc);

        // TODO 2.连接两条流
        BroadcastConnectedStream<String, String> inputControlBCS = inputDS.connect(controlBS);

        // TODO 3.调用算子进行处理
        inputControlBCS
                .process(new MyBroadcastProcessFunction())
                .print();


        env.execute();
    }

    public static class MyBroadcastProcessFunction extends BroadcastProcessFunction<String, String, String> {

        /**
         * 处理 非广播流 的数据
         *
         * @param value
         * @param ctx
         * @param out
         * @throws Exception
         */
        @Override
        public void processElement(String value, ReadOnlyContext ctx, Collector<String> out) throws Exception {
            // TODO 5.根据 广播状态(只读)的值，进行处理
            ReadOnlyBroadcastState<String, String> broadcastState = ctx.getBroadcastState(broadcastStateDesc);
            String aSwitch = broadcastState.get("switch");
            if ("1".equals(aSwitch)) {
                out.collect("广播流的数据为1，切换成 1 的执行逻辑");
            } else if ("0".equals(aSwitch)) {
                out.collect("广播流的数据为0，切换成 0 的执行逻辑");
            } else {
                out.collect("广播流的数据为其他，切换成 其他 执行逻辑");
            }
        }

        /**
         * 处理 广播流 的数据
         *
         * @param value
         * @param ctx
         * @param out
         * @throws Exception
         */
        @Override
        public void processBroadcastElement(String value, Context ctx, Collector<String> out) throws Exception {
            //TODO 4.对 广播状态 进行赋值
            BroadcastState<String, String> broadcastState = ctx.getBroadcastState(broadcastStateDesc);
            broadcastState.put("switch", value);
        }
    }

}
/*
    广播状态的适用场景
        1、广播出去的流，应该是 数据量小、变化缓慢 的输入
        2、动态开关（执行不同逻辑）、动态配置更新
 */